package com.at.checkpoint15;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 保存点
 *
 * @author huangchao E-mail:fengquan8866@163.com
 * @version 创建时间：2024/10/3 21:23
 */
public class SavepointDemo2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 导入hadoop依赖，指定访问hdfs的用户名
        System.setProperty("HADOOP_USER_NAME", "root");

        // 1、启用检查点
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 2、指定检查点的存储位置
        checkpointConfig.setCheckpointStorage("hdfs://master:8020/flink/checkpoint");
        // 3、checkpoint超时时间: 默认10分钟
        checkpointConfig.setCheckpointTimeout(60*1000);
        // 4、同时运行中的checkpoint的最大数量
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        // 5、最小等待时间: 上一轮checkpoint结束 到 下一轮 checkpoint开始 之间的间隔，设置了>0，并发就会变成1
        checkpointConfig.setMinPauseBetweenCheckpoints(1000);
        // 6、取消作业时，checkpoint的数据保留在外部系统 是否保留在外部系统
        // DELETE_ON_CANCELLATION: 主动cancel时，删除存在外部系统的 chk-xx 目录（如果是程序突然挂掉，不会删）
        // RETAIN_ON_CANCELLATION：主动cancel时，外部系统的 chk-xx 目录会保存下来
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 7、允许 checkpoint 连续失败的次数，默认0--》 表示checkpoint一失败，Job就挂掉
        checkpointConfig.setTolerableCheckpointFailureNumber(10);

        // 开启之后的要求：Checkpoint模式必须是精准一次，最大并发必须设为1
        checkpointConfig.enableUnalignedCheckpoints();
        // 开启非对齐检查点才生效：默认0，表示一开始就直接用 非对齐检查点
        // 如果>0，一开始用 对齐检查点（Barrier对齐），对齐的时间超过这个参数，自动切换成 非对齐检查点（Barrier非对齐）
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(1));

        env.socketTextStream("master", 7777).uid("socket")
                .flatMap(
                        (String value, Collector<Tuple2<String, Integer>> out) -> {
                            String[] words = value.split(" ");
                            for (String word : words) {
                                out.collect(Tuple2.of(word, 1));
                            }
                        }).uid("flatmap-wc").name("wc-flatmap")
//                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .returns(new TypeHint<Tuple2<String, Integer>>() {
                })
                .keyBy(value -> value.f0)
                .sum(1).uid("sum-wc")
                .print().uid("print-wc");

        env.execute();
    }
}
